package cn.itcast.spark.sql

import org.apache.spark.sql.SparkSession
import org.junit.Test

class JoinProcessor {

  val spark = SparkSession.builder()
    .master("local[6]")
    .appName("agg")
    .getOrCreate()

  import spark.implicits._
  import org.apache.spark.sql.functions._

  val person = Seq((0, "Lucy", 0), (1, "Lily", 0), (2, "Tim", 2), (3, "Danial", 3))
    .toDF("id", "name", "cityId")

  person.createOrReplaceTempView("person")

  val cities = Seq((0, "Beijing"), (1, "Shanghai"), (2, "Guangzhou"))
    .toDF("id", "name")

  cities.createOrReplaceTempView("cities")

  @Test
  def introJoin(): Unit = {

    val person = Seq((0, "Lucy", 0), (1, "Lily", 0), (2, "Tim", 2), (3, "Danial", 0))
      .toDF("id", "name", "cityId")

    val cities = Seq((0, "Beijing"), (1, "Shanghai"), (2, "Guangzhou"))
      .toDF("id", "name")

    val df = person.join(cities, person.col("cityId") === cities.col("id"))
      .select(person.col("id"), person.col("name"), cities.col("name") as "city")
//      .show()

    df.createOrReplaceTempView("user_city")

    spark.sql("select id, name, city from user_city where city = 'Beijing'")
      .show()
  }

  @Test
  def crossJoin(): Unit ={
    person.crossJoin(cities)
      .where(person.col("cityId") === cities.col("id"))
      .show()

    spark.sql("select u.id, u.name, c.name " +
      "from person u cross join cities c where u.cityId = c.id")
      .show()
  }

  @Test
  def inner(): Unit ={
    person.join(cities, person.col("cityId") === cities.col("id"), joinType = "inner")
      .show()

    spark.sql("select p.id, p.name, c.name " +
      "from person p inner join cities c " +
      "on p.cityId = c.id")
      .show()
  }

  @Test
  def fullOuter(): Unit ={
    person.join(cities, person.col("cityId") === cities.col("id"), joinType = "full")
      .show()

    spark.sql("select p.id, p.name, c.name " +
      "from person p full join cities c " +
      "on p.cityId = c.id")
      .show()
  }

  @Test
  def leftOuter(): Unit ={
    person.join(cities, person.col("cityId") === cities.col("id"), joinType = "left")
      .show()

    spark.sql("select p.id, p.name, c.name " +
      "from person p left join cities c " +
      "on p.cityId = c.id")
      .show()
  }

}










